package com.atguigu.ordertimeout_detect

import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.Map

/**
 * 订单支付实时监控
 * 使用CEP实现对创建和支付订单的核对
 *
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.ordertimeout_detect
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/15 10:17
 */
object OrderTimeout {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val path: String = OrderTimeout.getClass.getClassLoader.getResource("OrderLog.csv").getPath

    val value = env.readTextFile(path)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")

          OrderEvent(dataArrays(0).toLong, dataArrays(1), dataArrays(2), dataArrays(3).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(1)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })
      .keyBy(_.orderId)

    //定义CEP过滤条件
    val pattern: Pattern[OrderEvent, OrderEvent] = Pattern.begin[OrderEvent]("start").where(_.eventType == "create")
      .followedBy("follow").where(_.eventType == "pay")
      .within(Time.minutes(15))

    //使用CEP过滤出15分钟的超时订单和支付成功订单
    val cepPattern = CEP.pattern(value, pattern)
      .flatSelect(new OutputTag[WarningOreder]("no pay timeout"))(
        (map: Map[String, Iterable[OrderEvent]], time: Long, out: Collector[WarningOreder]) => {
          val head = map.get("start").get.head
          out.collect(WarningOreder(head.orderId, head.eventTime, "no pay timeout"))
        })((map: Map[String, Iterable[OrderEvent]], out: Collector[OrderEvent]) => {
        val head: OrderEvent = map.get("follow").get.head
        out.collect(head)
      })

    //输出超时订单到测输出流，主流输出正常订单
    cepPattern.getSideOutput(new OutputTag[WarningOreder]("no pay timeout")).print("sideOutput").setParallelism(1)
    cepPattern.print("ouput").setParallelism(1)

    env.execute("OrderTimeout job")
  }
}


case class OrderEvent(
                       orderId: Long,
                       eventType: String,
                       pyId: String,
                       eventTime: Long
                     )

case class WarningOreder(
                          orderId: Long,
                          eventTime: Long,
                          msg: String
                        )

